package pack01;

import org.apache.commons.compress.utils.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.File;
import java.io.FileInputStream;
import java.net.URI;

/**
 * @author lilulu
 * @date 2023/2/9 23:48
 */
public class TestDemo01HDFS2 {
    private FileSystem fileSystem;

    @Before
    public void init() throws Exception {
        //获取FileSystem对象
        fileSystem = FileSystem.get(new URI("hdfs://node1:8020"), new Configuration());
    }

    @After
    public void close() throws Exception {
        //关闭文件系统
        fileSystem.close();
    }


    //2:在HDFS上创建目录
    @Test
    public void mkdirTest() throws Exception {
        //创建目录
        fileSystem.mkdirs(new Path("/xxx/yyy/zzz")); //写权限

    }

    //3:删除目录
    @Test
    public void deleteTest() throws Exception {
        //删除目录
        fileSystem.delete(new Path("/xxx"), true);
    }

    //4:实现 hadoop fs -ls /功能，列出指定目录下的文件和文件夹
    @Test
    public void listFiles() throws Exception {

        //1:获取/目录下所有文件的详情，并将详情信息存入数组
        FileStatus[] fileStatuses = fileSystem.listStatus(new Path("/dir"));

        //2：遍历数组，获取每一个文件的具体详情
        for (FileStatus fileStatus : fileStatuses) {
            System.out.println(fileStatus.getPath()); //获取文件的路径
            System.out.println(fileStatus.getBlockSize()); //获取block大小
            System.out.println(fileStatus.getReplication()); //获取副本数
            System.out.println(fileStatus.getLen()); //获取文件的长度
            System.out.println("---------------------------------");

            //获取每一个文件的Block信息
            LocatedFileStatus locatedFileStatus = (LocatedFileStatus)fileStatus;

            //获取文件的每一个Block详情
            BlockLocation[] blockLocations = locatedFileStatus.getBlockLocations();
            System.out.println(blockLocations.length);
        }
    }


    //5:实现 hadoop fs -ls /功能，列出指定目录下的非文件夹文件
    @Test
    public void listFiles2() throws Exception {

        //获取RemoteIterator 得到所有的文件或者文件夹，第一个参数指定遍历的路径，第二个参数表示是否要递归遍历
        RemoteIterator<LocatedFileStatus> locatedFileStatusRemoteIterator =
                fileSystem.listFiles(new Path("/dir"), true);

        //遍历集合
        while (locatedFileStatusRemoteIterator.hasNext()){

            LocatedFileStatus fileStatus = locatedFileStatusRemoteIterator.next();
            System.out.println(fileStatus.getPath());

            //获取文件的Block信息
            BlockLocation[] blockLocations = fileStatus.getBlockLocations(); //该数组包含了文件每一个BLock信息
            System.out.println(blockLocations.length); //数组的长度就是每一个文件的Block数
            for (BlockLocation blockLocation : blockLocations) {  //遍历每一个Block
                String[] hosts = blockLocation.getHosts(); //获取Block的3个副本所在的主机信息
                System.out.println("副本数："+hosts.length);
                for (String host : hosts) { //获取每一个副本所在的主机
                    System.out.println(host);
                }
                System.out.println("***********");
            }
        }

    }


    //6:从HDFS上下载文件
    @Test
    public void downloadFile() throws Exception {
        fileSystem.copyToLocalFile(new Path("/dir/a.txt"),new Path("D:\\"));
    }


    //6:从HDFS上传文件
    @Test
    public void uploadFile() throws Exception {
        //fileSystem.copyFromLocalFile(new Path("D:\\a.txt"),new Path("/dir/aaa.txt"));


        //获取源数据的输入流
        FileInputStream fileInputStream = new FileInputStream("D:\\a.txt");

        //获取目标文件的输出流

        FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path("/dir/bbb.txt"));

        //实现文件的复制
        byte[] bytes = new byte[1024];

        while (true){
            int len = fileInputStream.read(bytes); //从源文件读取数据，返回实际读取的字节数
            if(len == -1){
                break;
            }
            fsDataOutputStream.write(bytes,0,len); //将读取到的数据写到目标文件
        }
        fsDataOutputStream.close();
        fileInputStream.close();
    }


    //7:实现小文件合并上传  hadoop fs -appendToFile
    @Test
    public void appendToFile() throws Exception {
        //获取源数据的输入流
        File file = new File("E:\\input");
        File[] files = file.listFiles();

        //获取目标文件的输出流
        FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path("/dir/big_file.txt"));

        for (File file1 : files) {
            //获取每一个文件的输入流
            FileInputStream fileInputStream = new FileInputStream(file1);

            //将每一个小文件追击到目标文件中
            IOUtils.copy(fileInputStream,fsDataOutputStream); //追加复制
        }
        fsDataOutputStream.close();
    }
}
